package com.atguigu.edu.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.util.KafkaUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.ZoneId;

/**
 * @author zzw
 * @create 2022/9/4 9:28
 */
public class DwdVideoLookProcess {
    public static void main(String[] args) {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        env.setStateBackend(new HashMapStateBackend());
        */

        // 设置时区的方法
        tableEnv.getConfig().setLocalTimeZone(ZoneId.of("GMT+8"));

        // TODO 3 读取kafka主题数据  dwd_traffic_app_Video
        String topicName = "dwd_traffic_app_Video";
        String grooupId = "dwd_video_look_process";
        DataStreamSource<String> processStream = env.addSource(KafkaUtil.getKafkaConsumer(topicName, grooupId));

        // TODO 4 转换数据格式
        SingleOutputStreamOperator<JSONObject> jsonObjStream = processStream.map(JSON::parseObject);

        // TODO 5 写出到新的卡夫卡主题中  dwd_video_look_process
        String targetTopic = "dwd_video_look_process";

        processStream.addSink(KafkaUtil.getKafkaProducer(targetTopic));
        jsonObjStream.map(JSONAware::toJSONString).addSink(KafkaUtil.getKafkaProducer(targetTopic));
    }
}
